
package cn.junechiu.learnrxjava;

import android.os.Bundle;
import android.os.SystemClock;
import android.support.annotation.Nullable;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.widget.Toast;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

/**
 * Created by android on 2017/7/19. RxJava 2.0
 * 最核心的是Publisher和Subscriber。Publisher可以发出一系列的事件，而Subscriber负责和处理这些事件。
 * 平常用得最多的Publisher是Flowable，它支持背压
 */
public class FlowableActivity extends AppCompatActivity {

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
//        sample1();
//        sample2();
//        sample3();
//        sample4();
//        sample5();
//        sample6();
//        sample7();
        sample8();
        sample9();
    }

    /**
     * 在onSubscribe中，我们需要调用request去请求资源，参数就是要请求的数量，一般如果不限制请求数量，
     * 可以写成Long.MAX_VALUE。 如果你不调用request，Subscriber的onNext和onComplete方法将不会被调用。
     */
    public void sample1() {
        Flowable flowable = Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> e) throws Exception {
                e.onNext("ddddd");
                e.onComplete();
            }
        }, BackpressureStrategy.BUFFER); // 缓存背压

        Subscriber subscriber = new Subscriber() {
            @Override
            public void onSubscribe(Subscription s) {
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Object o) {
                Log.d("FlowableActivity", "sample1: " + o.toString());
            }

            @Override
            public void onError(Throwable t) {
            }

            @Override
            public void onComplete() {
            }
        };
        flowable.subscribe(subscriber);
    }

    public void sample2() {
        Flowable.just("hello").subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d("FlowableActivity", "sample2: " + s.toString());
            }
        });
    }

    public void sample3() {
        Flowable.just("hello world").map(new Function<String, Integer>() {   //string --> int
            @Override
            public Integer apply(String s) throws Exception {
                return s.hashCode();
            }
        }).map(new Function<Integer, String>() { //int --> string
            @Override
            public String apply(Integer integer) throws Exception {
                return String.valueOf(integer);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d("FlowableActivity", "sample3: " + s.toString());
            }
        });
    }

    public void sample4() {
        List<Integer> list = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            list.add(i);
        }
        //遍历发射数组中的数据
        Flowable.fromIterable(list).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d("FlowableActivity", "sample4: " + integer);
            }
        });

        //just
        Flowable.just(list)
                .flatMap(new Function<List<Integer>, Publisher<Integer>>() { //list-->Publisher
                    @Override
                    public Publisher<Integer> apply(List<Integer> integers) throws Exception {
                        return Flowable.fromIterable(integers);
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d("FlowableActivity", "sample4--2: " + integer);
            }
        });
    }

    /**
     * filter 是用于过滤数据的，返回false表示拦截此数据。
     */
    public void sample5() {
        Flowable.fromArray(1, 4, 5, 6, 2, 3).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer > 5;  //过滤大于5的数据
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d("FlowableActivity", "sample5: " + integer);
            }
        });
    }

    //take 用于指定订阅者最多收到多少数据
    public void sample6() {
        Flowable.fromArray(1, 2, 3, 4, 5, 6, 7).take(3).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d("FlowableActivity", "sample6: " + integer);
            }
        });
    }

    //doOnNext 允许我们在每次输出一个元素之前做一些额外的事情
    public void sample7() {
        Flowable.fromArray(1, 2, 3, 4).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d("FlowableActivity", "sample7--保存: " + integer);
            }
        }).doOnSubscribe(new Consumer<Subscription>() {
            @Override
            public void accept(Subscription subscription) throws Exception {
                Log.d("FlowableActivity", "sample7--订阅前: " + subscription);
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d("FlowableActivity", "sample7: " + integer);
            }
        });
    }

    public void sample8() {
        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> e) throws Exception {
                e.onNext("3秒后显示");
                SystemClock.sleep(3000);
                e.onNext("haha");
            }
        }, BackpressureStrategy.BUFFER)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Toast.makeText(FlowableActivity.this, s, Toast.LENGTH_LONG).show();
                    }
                });
    }

    //把发射的数据获取出来而不是通过订阅者来输出
    public void sample9() {
        List<String> list = Flowable.range(1, 100).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "hhh" + integer;
            }
        }).toList().blockingGet();
        System.out.println("FlowableActivity" + list);
    }
}
